package com.lagou.numbersort.step3;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.*;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

public class NumSortMapper3 extends Mapper<LongWritable, Text, LongWritable, Text> {

    private LongWritable k = new LongWritable();
    private Map<String,Long> partCountMap = new HashMap<>();
    private Long startIndex = 1L;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {

        URI[] cacheFiles = context.getCacheFiles();
        String path = cacheFiles[0].getPath().toString();
        //读取缓存文件内容（每个分区的名称及对应的数字个数）
        BufferedReader reader = new BufferedReader(new InputStreamReader(
                new FileInputStream(path), "utf-8"));
        String line = "";
        while(StringUtils.isNotEmpty(line = reader.readLine())){
            String[] splits = line.split("\t");
            partCountMap.put(splits[0],Long.parseLong(splits[1]));
        }
        reader.close();

        InputSplit inputSplit = context.getInputSplit();
        FileSplit split = (FileSplit) inputSplit;

        String reduceNumStr = split.getPath().getName().split("-")[2];
        int reduceNum = Integer.parseInt(reduceNumStr);
        //获取每个分区中数字的其实index
        for (int i = 0; i < reduceNum; i++) {
            String strRedeceName = "part-r-" + getReduceTaskResultName(i);
            startIndex += partCountMap.get(strRedeceName);
        }
    }

    private String getReduceTaskResultName(int i) {
        if (i < 10){
            return "0000" + i;
        }else if (i < 100){
            return "000" + i;
        }else if (i < 1000){
            return "00" + i;
        }else  if (i < 10000){
            return "0" + i;
        }else {
            return String.valueOf(i);
        }
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws
            IOException, InterruptedException {

        k.set(startIndex);
        context.write(k,value);
        startIndex++;//序号自增
    }
}
